

package com.coding4play.mqtt.broker.internal;


import com.coding4play.mqtt.broker.config.BrokerProperties;
import com.coding4play.mqtt.broker.service.KafkaService;
import com.coding4play.mqtt.broker.util.JdkTool;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;


/**
 * 聚合消息消费端
 * kafka
 * redis（单机 ｜ 集群）
 */
@Component
@AllArgsConstructor
@Slf4j
public class AggregatedSink {

    private BrokerProperties brokerProperties;
    private KafkaService kafkaService;
    private RedisTemplate redisTemplate;

    public void sink(InternalMessage internalMessage) {
        String processId = JdkTool.getProcessId("0");
        //broker唯一标识 mqttwk.broker.id
        internalMessage.setBrokerId(brokerProperties.getId());
        internalMessage.setProcessId(processId);
        //如果开启kafka消息转发
        if (brokerProperties.getKafka().getBrokerEnabled()) {
            kafkaService.send(internalMessage);
        }
        //如果开启集群功能
        if (brokerProperties.getClusterEnabled()) {
            redisTemplate.convertAndSend(RedisListener.CLUSTER_TOPIC, internalMessage);
        }
    }
}
